package com.sxyjhh.service.data_exchange;

import com.robert.kafka.kclient.core.KafkaProducer;
import com.sxyjhh.bean.AqDataPushRecode;
import com.sxyjhh.common.DBConfig;
import com.sxyjhh.create.api.DataApi;
import com.sxyjhh.dao.AqDataPushRecodeRepository;
import com.sxyjhh.dbutil.basic.ConstConfig;
import com.sxyjhh.dbutil.basic.DBQueryUtils;
import com.sxyjhh.dbutil.utils.CommonUtils;
import com.sxyjhh.dbutil.utils.DBInfoUtil;
import com.sxyjhh.dbutil.utils.PageInfo;
import com.sxyjhh.schedule.TaskSchedulerManager;
import com.sxyjhh.utils.IdGenerator;
import com.sxyjhh.utils.redis.RedisOperator;
import com.sxyjhh.utils.util.Md5Util;
import com.sxyjhh.utils.util.XaUtil;
import net.sf.json.JSONArray;
import net.sf.json.JSONObject;
import org.apache.commons.collections.map.HashedMap;
import org.apache.commons.lang.StringUtils;
import org.apache.http.client.utils.CloneUtils;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;
import redis.clients.jedis.JedisCluster;

import java.sql.SQLException;
import java.util.*;

/**
 * Created by spring on 2018/6/12.
 */
@Component
public class DataExchangeService {

    @Autowired
    TaskSchedulerManager taskSchedulerManager;

    /* @Autowired
     private KafkaProducer fafkaProducer;
 */
    private Map<String, String> dicMap = new HashMap<>();

    private String ip = DBConfig.readPop("output.datasource.ip");
    private String dbName = DBConfig.readPop("output.datasource.db.name");
    private String userName = DBConfig.readPop("output.datasource.username");
    private String password = DBConfig.readPop("output.datasource.password");
    private String port = DBConfig.readPop("output.datasource.port");
    private String dbSType = DBConfig.readPop("output.datasource.type");
    private String dicTableName = DBConfig.readPop("DIC.TABNAME");


    private String recordIp = DBConfig.readPop("output.record.ip");
    private String recordDbName = DBConfig.readPop("output.record.db.name");
    private String recordUserName = DBConfig.readPop("output.record.username");
    private String recordPassword = DBConfig.readPop("output.record.password");
    private String recordPort = DBConfig.readPop("output.record.port");
    private String recordDbSType = DBConfig.readPop("output.record.type");
    private static final Logger log = Logger.getLogger(DataExchangeService.class);

    @Autowired
    DBQueryUtils dbQueryUtils;

    @Autowired
    private JedisCluster jedisCluster;

    @Autowired
    private RedisOperator redisOperator;

    @Autowired
    private AqDataPushRecodeRepository aqDataPushRecodeRepository;

    @Bean
    public KafkaProducer getKafkaProducer() {
        return new KafkaProducer(
                "kafka-producer.properties", "test");
    }

    @Bean
    public DBQueryUtils getDBQueryUtils() {
        String url = "";
        ConstConfig.DBType dbType = ConstConfig.DBType.valueOf(Integer.valueOf(dbSType));
        if ("3".equals(dbSType)) {
            url = "jdbc:mysql://" + ip + ":" + port + "/" + dbName + "?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useInformationSchema=true&useSSL=false";
        } else {
            url = DBInfoUtil.getDBUrl(dbType, ip, port, dbName);
        }

        return new DBQueryUtils(DBInfoUtil.getDBDriverName(dbType),
                url, userName, password, 5);
    }

    @Autowired
    PushService pushService;

    //获取数据
    public void getData(String tabName, String outTabName, String key, DataApi ri, int pushNum, String fType) throws SQLException {

        Date sTime = new Date();

        String condition = "";
        String type = "";

        if ("ag_ry".equals(fType)) {
            condition = " AND FPersonTypeId in (60004,60005,60006,60007,60008,60009)";
            type = "PERSON";
        }
        if ("tz_zy".equals(fType)) {
            condition = " AND FPersonTypeId in (21501,21502,21503,21504,21505,21506,21507,21508,21509,21510,21511) AND  FSTATE IN (-1,1) ";
            type = "SPECIAL";
        }
        if ("axz".equals(fType)) {
            type = "SECURITY";
        }

        // 检测hbase 表数据条数
        String countSql = "SELECT count(1) from " + tabName + " where 1=1 " + condition;

        List<String> dataList = ri.selectDataQ(countSql);
        if (XaUtil.isEmpty(dataList) || dataList.size() < 1 || "0".equals(dataList.get(0))) {
            return;
        }

        int count = Integer.parseInt(dataList.get(0));

        int num = 3000;
        int batch = (count / num) + (count % num != 0 ? 1 : 0);

        int outCount = 0;
        List countList = dbQueryUtils.queryData("select count(1) as CUT from " + outTabName);
        if (!XaUtil.isEmpty(countList)) {
            JSONObject countJson = JSONObject.fromObject(countList.get(0));
            outCount = Integer.parseInt(countJson.get("CUT") + "");
        }

        List<String> dicList = ri.selectDataQ("select FNUMBER,FNAME from " + dicTableName);
        if (!XaUtil.isEmpty(dicList)) {
            for (String dic : dicList) {
                String[] dicArray = dic.split("&&");
                dicMap.put(dicArray[0], dicArray[1]);
            }
        }

        AqDataPushRecode aqDataPushRecode = new AqDataPushRecode();

        aqDataPushRecode.setDataType(type);

        aqDataPushRecode.setNum(0);
        aqDataPushRecode.setTotalNum(0);

        aqDataPushRecodeRepository.save(aqDataPushRecode);

        if (outCount > 0) {
            int b = cacheData(outTabName, outCount, dbQueryUtils, key, 0, 0);
            if (b == -1) {
                return;
            }
        }

        for (int i = 0; i < batch; i++) {
            log.info("==========数据对比批次==========");
            log.info((i == 0 ? 0 : i * num) + "");

            String outName = outTabName;
            List<Map<String, Object>> outList = new ArrayList<>();
            Map<String, Map<String, Map<String, Object>>> recordMap = getTableData(tabName, num, i == 0 ? 0 : i * num, ri, key, condition);
            dataCompare(recordMap, tabName, dbQueryUtils, outCount, key, outList, outName);
            try {
                if (outList.size() > 0) {
                    log.info("本次推送数据量：" + outList.size());
                    conversionData(outName, outList, "1", dbQueryUtils, key, pushNum, aqDataPushRecode,fType);
                }
            } catch (CloneNotSupportedException e) {
                e.printStackTrace();
            }
        }

        //推送删除数据
       /* List<Map<String, Object>> delMap = new ArrayList<>();

        deleteData(outTabName, dbQueryUtils, key, delMap, ri, fType, aqDataPushRecode.getId());*/

        long endTime = new Date().getTime() - sTime.getTime();
        long finTime = (endTime % (60 * 60 * 1000)) / (60 * 1000);
        log.info("==========" + tabName + "==========");
        log.info("==========任务执行完成==========");
        log.info("==========任务耗时==========");
        log.info(finTime + "分钟");
        AqDataPushRecode dataPushRecode = aqDataPushRecodeRepository.findOne(aqDataPushRecode.getId());
        dataPushRecode.setEndTime(new Date());
        dataPushRecode.setFinishedTime(finTime);
        aqDataPushRecodeRepository.save(dataPushRecode);
    }

    public void dataCompareDel(String tabName, String outTabName, String key, DataApi ri, int pushNum, String fType) throws SQLException {

        Date sTime = new Date();

        String condition = "";
        String type = "";
        if ("ag_ry".equals(fType)) {
            type = "PERSON";
            condition = " AND FPersonTypeId in (60004,60005,60006,60007,60008,60009) ";
        }
        if ("tz_zy".equals(fType)) {
            type = "SPECIAL";
            condition = " AND FPersonTypeId in (21501,21502,21503,21504,21505,21506,21507,21508,21509,21510,21511) AND  FSTATE IN (-1,1) ";
        }
        if ("axz".equals(fType)) {
            type = "SECURITY";
        }

        // 检测hbase 表数据条数
        String countSql = "SELECT count(1) from " + tabName + " where 1=1 " + condition;

        List<String> dataList = ri.selectDataQ(countSql);
        if (XaUtil.isEmpty(dataList) || dataList.size() < 1 || "0".equals(dataList.get(0))) {
            return;
        }

        int count = Integer.parseInt(dataList.get(0));

        int num = 3000;
        int batch = (count / num) + (count % num != 0 ? 1 : 0);

        int outCount = 0;
        List countList = dbQueryUtils.queryData("select count(1) as CUT from " + outTabName);
        if (!XaUtil.isEmpty(countList)) {
            JSONObject countJson = JSONObject.fromObject(countList.get(0));
            outCount = Integer.parseInt(countJson.get("CUT") + "");
        }

        List<String> dicList = ri.selectDataQ("select FNUMBER,FNAME from " + dicTableName);
        if (!XaUtil.isEmpty(dicList)) {
            for (String dic : dicList) {
                String[] dicArray = dic.split("&&");
                dicMap.put(dicArray[0], dicArray[1]);
            }
        }

        AqDataPushRecode aqDataPushRecode = new AqDataPushRecode();

        aqDataPushRecode.setDataType(type);

        aqDataPushRecode.setNum(0);
        aqDataPushRecode.setTotalNum(0);

        aqDataPushRecodeRepository.save(aqDataPushRecode);

        if (outCount > 0) {
            int b = cacheData(outTabName, outCount, dbQueryUtils, key, 0, 0);
            if (b == -1) {
                return;
            }
        }

        for (int i = 0; i < batch; i++) {
            log.info("==========数据对比批次==========");
            log.info((i == 0 ? 0 : i * num) + "");

            String outName = outTabName;
            List<Map<String, Object>> outList = new ArrayList<>();
            Map<String, Map<String, Map<String, Object>>> recordMap = getTableData(tabName, num, i == 0 ? 0 : i * num, ri, key, condition);
            dataCompare(recordMap, tabName, dbQueryUtils, outCount, key, outList, outName);
        }

        //推送删除数据
        List<Map<String, Object>> delMap = new ArrayList<>();

        deleteData(outTabName, dbQueryUtils, key, delMap, ri, fType, aqDataPushRecode.getId());


        long endTime = new Date().getTime() - sTime.getTime();
        long finTime = (endTime % (60 * 60 * 1000)) / (60 * 1000);
        log.info("==========" + tabName + "==========");
        log.info("==========任务执行完成==========");
        log.info("==========任务耗时==========");
        log.info(finTime + "分钟");
        AqDataPushRecode dataPushRecode = aqDataPushRecodeRepository.findOne(aqDataPushRecode.getId());
        dataPushRecode.setEndTime(new Date());
        dataPushRecode.setFinishedTime(finTime);
        aqDataPushRecodeRepository.save(dataPushRecode);
    }

    //对比删除数据
    private void deleteData(String tabName, DBQueryUtils dbQueryUtils,
                            String key, List<Map<String, Object>> delMap,
                            DataApi ri, String type, String recodeId) {

        //删除记录
        Set<String> set = redisOperator.keys(tabName + "*");
        List<String> keys = new ArrayList<>();
        Map<String, Map<String, Object>> dMap = null;
        if (!XaUtil.isEmpty(set)) {
            dMap = new HashMap<>();
            for (String redisKey : set) {
                String json = jedisCluster.get(redisKey);
                Map jsMap = JSONObject.fromObject(json);
                jsMap.put("OPERATION_FLAG", "D");
                dMap.put(redisKey, jsMap);
                keys.add(redisKey);
            }
        }

        if (keys.size() <= 0) {
            return;
        }

        AqDataPushRecode aqDataPushRecode = aqDataPushRecodeRepository.findOne(recodeId);

        String daqTable = "";
        if (tabName.toUpperCase().contains("CF_ENT_SAFETYCERTI")) {
            daqTable = DBConfig.readPop("COMPANY.SAFETY.CERTIFICATE");
        }
        if (tabName.toUpperCase().contains("CF_PERSON_BASEINFO")) {
            daqTable = DBConfig.readPop("PERSON.SECURITY.CERTIFICATE");
        }

        int count = keys.size();
        int num = 1000;
        int batch = (count / num) + (count % num != 0 ? 1 : 0);
        StringBuilder sql = new StringBuilder();
        StringBuilder delSql = new StringBuilder();
        for (int i = 0; i < batch; i++) {
            sql.delete(0, sql.length());
            delSql.delete(0, delSql.length());
            delMap.clear();
            sql.append("SELECT " + key + " from ");
            sql.append(daqTable);
            sql.append(" where " + key + " IN ( '");
            List<String> keyPage = keys.subList(i * num, ((i + 1) * num) > count ? count : (i + 1) * num);
            for (int j = 0; j < keyPage.size(); j++) {
                if (j != 0) {
                    sql.append("','");
                }
                sql.append(keyPage.get(j).replace(tabName + "_", ""));
            }
            sql.append("')");
            if ("ag_ry".equals(type)) {
                sql.append(" AND FPersonTypeId in (60004,60005,60006,60007,60008,60009) ");
            }
            if ("tz_zy".equals(type)) {
                sql.append(" AND FPersonTypeId in (21501,21502,21503,21504,21505,21506,21507,21508,21509,21510,21511)  AND  FSTATE in (-1,1) ");
            }
            try {
                List<String> dataList = ri.selectDataQ(sql.toString());
                if (!XaUtil.isEmpty(dataList)) {
                    dataList.forEach(s -> {
                        keyPage.remove(s);
                    });
                }
                for (String s : keyPage) {
                    delMap.add(dMap.get(s));
                }

                if ("axz".equals(type)) {
                    delMap = pushService.pushZj(tabName, delMap);
                }
                if ("ag_ry".equals(type)) {
                    delMap = pushService.pushRy(tabName, delMap);
                }

                //删除对比表数据
                delSql.append(" delete from  ");
                delSql.append(tabName);
                delSql.append(" where 1=1 and ");
                delSql.append(key);
                delSql.append(" IN ( ");

                String colSql = "";
                for (int j = 0; j < delMap.size(); j++) {
                    if (j == 0) {
                        colSql += "'" + delMap.get(j).get(key) + "'";
                    } else {
                        colSql += ",'" + delMap.get(j).get(key) + "'";
                    }

                }
                delSql.append(colSql);
                delSql.append(" )");

                System.out.println("删除对比数据批次: " + i);
                System.out.println("删除对比数据数量: " + keyPage.size());
                dbQueryUtils.execDDLSql(delSql.toString());

            } catch (SQLException e1) {
                log.error(e1.getMessage());
                keys.forEach(s -> jedisCluster.del(s));
                return;
            }
            aqDataPushRecode.setDeleteNum(aqDataPushRecode.getDeleteNum() + delMap.size());
        }

        keys.forEach(s -> {
            jedisCluster.del(s);
        });
        keys.clear();
        aqDataPushRecodeRepository.save(aqDataPushRecode);
    }

    //数据比对判断数据状态为 新增 修改
    public void dataCompare(Map<String, Map<String, Map<String, Object>>> map, String tableName, DBQueryUtils dbQueryUtils, int outCount, String key, List<Map<String, Object>> outList, String outTabName) {
        if (outCount > 0) {
            if (tableName.toUpperCase().contains("CF_ENT_SAFETYCERTI")) {
                for (Map.Entry<String, Map<String, Map<String, Object>>> m : map.entrySet()) {
                    String keyValue = outTabName + "_" + m.getKey();
                    Map<String, Object> insertMap = null;
                    Map<String, Map<String, Object>> subMap = m.getValue();
                    String json = jedisCluster.get(keyValue);
                    if (StringUtils.isBlank(json)) {
                        for (Map.Entry<String, Map<String, Object>> subM : subMap.entrySet()) {
                            insertMap = subM.getValue();
                            insertMap.put("OPERATION_FLAG", "I");
                            outList.add(insertMap);
                        }
                    } else {
                        Map jsMap = JSONObject.fromObject(json);
                        String oldMD5 = jsMap.get("MD5") + "";
                        for (Map.Entry<String, Map<String, Object>> subM : subMap.entrySet()) {

                            insertMap = subM.getValue();
                            String md5 = insertMap.get("MD5") + "";
                            if (!oldMD5.equals(md5)) {
                                insertMap.put("OPERATION_FLAG", "U");
                                outList.add(insertMap);
                            }
                        }
                        jedisCluster.del(keyValue);
                    }

                }

            } else if (tableName.toUpperCase().contains("CF_PERSON_BASEINFO")) {
                //todo 暂时用
                List<Map<String, Object>> updateMd5List = new ArrayList<>();
                for (Map.Entry<String, Map<String, Map<String, Object>>> m : map.entrySet()) {
                    String keyValue = outTabName + "_" + m.getKey();
                    Map<String, Object> insertMap = null;
                    Map<String, Map<String, Object>> subMap = m.getValue();
                    String json = jedisCluster.get(keyValue);
                    if (StringUtils.isBlank(json)) {
                        for (Map.Entry<String, Map<String, Object>> subM : subMap.entrySet()) {
                            insertMap = subM.getValue();
                            insertMap.put("OPERATION_FLAG", "I");
                            outList.add(insertMap);
                        }
                    } else {
                        Map jsMap = JSONObject.fromObject(json);
                        String oldMD5 = jsMap.get("MD5") + "";
                        for (Map.Entry<String, Map<String, Object>> subM : subMap.entrySet()) {

                            insertMap = subM.getValue();
                            String md5 = insertMap.get("MD5") + "";
                            if (!oldMD5.equals(md5)) {
                                String ryMd5 = getRYMd5(jsMap);
                                if (!ryMd5.equals(md5)) {
                                    insertMap.put("OPERATION_FLAG", "U");
                                    outList.add(insertMap);
                                } else {
                                    log.info("MD5计算错误的数据");
                                    log.info("主键值");
                                    log.info(insertMap.get("FID"));
                                    Map<String, Object> md5Map = new HashMap<>();
                                    md5Map.put("MD5", ryMd5);
                                    md5Map.put("FID", insertMap.get("FID"));
                                    updateMd5List.add(md5Map);
                                }
                            }
                        }
                        jedisCluster.del(keyValue);
                    }
                }
                //修改不同的Md5值
                if (!XaUtil.isEmpty(updateMd5List)) {
                    log.info("MD5计算错误的数据数量 ：" + updateMd5List.size());
                    updateData(outTabName, dbQueryUtils, updateMd5List, key, "1");
                }
            }
        } else {
            for (Map.Entry<String, Map<String, Map<String, Object>>> m : map.entrySet()) {
                Map<String, Object> insertMap = null;
                Map<String, Map<String, Object>> subMap = m.getValue();
                for (Map.Entry<String, Map<String, Object>> subM : subMap.entrySet()) {
                    insertMap = subM.getValue();
                    insertMap.put("OPERATION_FLAG", "I");
                }
                outList.add(insertMap);
            }
        }
    }

    //获取数据并计算字段MD5值
    public Map<String, Map<String, Map<String, Object>>> getTableData(String tableName, int num, int offsetS, DataApi ri, String key, String condition) throws SQLException {
        Map<String, Map<String, Map<String, Object>>> recordMap = new HashMap<>();

        StringBuffer sql = new StringBuffer();
        sql.append(" select * from ");
        sql.append(tableName);
        sql.append(" where 1=1 ");
        sql.append(condition);
        sql.append(" order by ");
        sql.append(key);
        sql.append(" limit " + num + " offset " + offsetS);
        List<String> dataList = ri.selectDataQ(sql.toString());
        List<String> colList = ri.getcolMeta(tableName);

        Map<String, String> colMap = new HashedMap();
        for (int j = 0; j < colList.size(); j++) {
            colMap.put(j + "", colList.get(j));
        }
        if (!XaUtil.isEmpty(dataList) && dataList.size() > 0) {
            for (int i = 0; i < dataList.size(); i++) {
                Map<String, Map<String, Object>> recordList = new HashMap<>();
                Map<String, Object> dataMap = new HashedMap();
                String[] dataArray = dataList.get(i).split("&&");
                String colValue = "";
                String keyValue = "";
                for (int j = 0; j < dataArray.length; j++) {
                    if("UUID".equals(colMap.get(j + ""))){
                        continue;
                    }
                    if("7f7c7211-c848-4be6-9f65-bcf655c98097".equals(dataArray[j])){

                        System.out.println(dataArray[j]);
                    }
                    colValue += dataArray[j];
                    dataMap.put(colMap.get(j + ""), "null".equals(dataArray[j]) ? "" : dataArray[j] + "");
                    if (key.equalsIgnoreCase(colMap.get(j + ""))) {
                        keyValue = dataArray[j];
                    }
                }
                String Md5Value = "";
                if (tableName.toUpperCase().contains("CF_PERSON_BASEINFO")) {
                    Md5Value = getRYMd5(dataMap);
                } else {
                    Md5Value = Md5Util.getMD5ofStr(colValue);
                }
                dataMap.put("MD5", Md5Value);
                recordList.put(Md5Value, dataMap);
                recordMap.put(keyValue, recordList);
            }
        }
        return recordMap;
    }

    private String getRYMd5(Map<String, Object> dataMap) {

        StringBuilder param = new StringBuilder();
        param.append(dataMap.get("FSEX") + "");
        param.append(dataMap.get("FID") + "");
        param.append(dataMap.get("FNAME") + "");
        param.append(dataMap.get("FIDCARDTYPEID") + "");
        param.append(dataMap.get("FIDCARD") + "");
        param.append(dataMap.get("FBIRTHDAY") + "");
        param.append(dataMap.get("FDEGREEID") + "");
        param.append(dataMap.get("FPERSONTYPEID") + "");
        param.append(dataMap.get("FBASENAME") + "");
        param.append(dataMap.get("FJURIDCIALCODE") + "");
        param.append(dataMap.get("FFUNCTION") + "");
        param.append(dataMap.get("FCERTINO") + "");
        param.append(dataMap.get("FBEGINTIME") + "");
        param.append(dataMap.get("FENDTIME") + "");
        param.append(dataMap.get("FAPPTIME") + "");
        param.append(dataMap.get("FSTATE") + "");

        String md5 = Md5Util.getMD5ofStr(param.toString());

        return md5;
    }

    //新增轮询表数据
    private void insertData(String tabName, DBQueryUtils dbQueryUtils, List<Map<String, Object>> resList, String type) {

        if (XaUtil.isEmpty(resList)) {
            return;
        }

        try {
            boolean flag = dbQueryUtils.insertBatchData(tabName, dbName, resList);
            if (!flag) {
                return;
            } else {
                log.info(tabName + "表 数据导入成功 :" + resList.size());
            }

        } catch (SQLException e) {
            e.printStackTrace();
            errorRecord(e.toString(), "sqlError", tabName, "insertData");
        }
    }

    //修改轮询表数据
    private void updateData(String tabName, DBQueryUtils dbQueryUtils, List<Map<String, Object>> resList, String key, String type) {

        if (XaUtil.isEmpty(resList)) {
            return;
        }

        List<String> sqls = new ArrayList<>();
        StringBuilder sql = new StringBuilder();
        for (int i = 0; i < resList.size(); i++) {
            sql.delete(0, sql.length());
            Map<String, Object> map = resList.get(i);

            if (tabName.toUpperCase().contains("CF_PERSON_BASEINFO")) {
                map.remove("FTXT1");
                map.remove("FTXT2");
                map.remove("FTXT3");
                map.remove("FTXT4");
                map.remove("FTXT5");
            }

            String keyValue = map.get(key) + "";
            sql.append("update ");
            sql.append(tabName);
            sql.append(" set ");
            String colSql = "";
            for (Map.Entry<String, Object> m : map.entrySet()) {
                Object obj = m.getValue();
                String objType = getObjType(obj);
                if ("3".equals(dbSType)) {
                    colSql += ",`" + m.getKey() + "` =  ";
                } else {
                    colSql += "," + m.getKey() + " =  ";
                }

                if ("int".equals(objType) || "Double".equals(objType) || "Float".equals(objType) || "Long".equals(objType)) {
                    colSql += obj;

                } else {
                    if (XaUtil.isEmpty(obj)) {
                        colSql += null;
                    } else {
                        colSql += " '" + obj + "' ";
                    }
                }

            }
            sql.append(colSql.substring(1, colSql.length()));
            sql.append(" where ");
            sql.append(key);
            sql.append(" = '");
            sql.append(keyValue);
            sql.append("' ");
            sqls.add(sql.toString());
        }

        String[] sqlArray = new String[]{};
        try {
            boolean flag = dbQueryUtils.execDDLSql(sqls.toArray(sqlArray));
            if (!flag) {
                return;
            } else {
                log.info(tabName + "表 数据修改成功 :" + resList.size());
            }
        } catch (SQLException e) {
            e.printStackTrace();
            errorRecord(e.toString() + "  sql语句 :" + JSONArray.fromObject(sqls).toString(), "sqlError", tabName, "updateData");
        }
    }

    private void conversionData(String tabName, List<Map<String, Object>> resList, String type,
                                DBQueryUtils dbQueryUtils, String key, int pushNum, AqDataPushRecode aqDataPushRecodeArg, String fType) throws CloneNotSupportedException {
        List<Map<String, Object>> insertList = null;
        List<Map<String, Object>> updateList = null;

        AqDataPushRecode aqDataPushRecode = aqDataPushRecodeRepository.findOne(aqDataPushRecodeArg.getId());
        long totalNum = aqDataPushRecode.getTotalNum();
        long num = aqDataPushRecode.getNum();
        long upNum = aqDataPushRecode.getUpdateNum();
        long inNum = aqDataPushRecode.getInsertNum();

        aqDataPushRecode.setTotalNum(totalNum + resList.size());

        if ("axz".equals(fType)) {
            resList = pushService.pushZj(tabName, resList);
        }
        if ("ag_ry".equals(fType)) {
            resList = pushService.pushRy(tabName, resList);
        }
        if ("tz_zy".equals(fType)) {
            resList = pushService.pushTz(tabName, resList);
        }

        if (XaUtil.isEmpty(resList)) {
            return;
        }
        aqDataPushRecode.setNum(num + resList.size());

        for (Map<String, Object> map : resList) {

            String operationFlag = map.get("OPERATION_FLAG") + "";
            if ("I".equals(operationFlag)) {
                if (XaUtil.isEmpty(insertList)) {
                    insertList = new ArrayList<>();
                }
                Map<String, Object> resMap = cloneMap(map);

                resMap.remove("OPERATION_FLAG");
                insertList.add(resMap);
            } else if ("U".equals(operationFlag)) {
                if (XaUtil.isEmpty(updateList)) {
                    updateList = new ArrayList<>();
                }
                Map<String, Object> resMap = (Map<String, Object>) CloneUtils.clone(map);
                resMap.remove("OPERATION_FLAG");
                updateList.add(resMap);
            }
        }

        if (!XaUtil.isEmpty(insertList)) {
            log.info("新增数据推送量：" + insertList.size());
            aqDataPushRecode.setInsertNum(inNum + insertList.size());
            insertData(tabName, dbQueryUtils, insertList, type);
        }

        if (!XaUtil.isEmpty(updateList)) {
            log.info("修改数据推送量：" + updateList.size());
            aqDataPushRecode.setUpdateNum(upNum + updateList.size());
            updateData(tabName, dbQueryUtils, updateList, key, type);
        }

        aqDataPushRecodeRepository.save(aqDataPushRecode);

    }

    public void startSynData() {
        DataApi ri = new DataApi(DBConfig.readPop("daq.task.pconnect.url"));

        String COMPANY_SAFETY_CERTIFICATE = DBConfig.readPop("COMPANY.SAFETY.CERTIFICATE");
        String COMPANY_SAFETY_CERTIFICATE_OUT = DBConfig.readPop("COMPANY.SAFETY.CERTIFICATE.OUT");

        try {
            getData(COMPANY_SAFETY_CERTIFICATE, COMPANY_SAFETY_CERTIFICATE_OUT, "FID", ri, 0, "axz");
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }

    public void startSynDataRy() {

        DataApi ri = new DataApi(DBConfig.readPop("daq.task.pconnect.url"));
        String PERSON_SECURITY_CERTIFICATE = DBConfig.readPop("PERSON.SECURITY.CERTIFICATE");
        String PERSON_SECURITY_CERTIFICATE_OUT = DBConfig.readPop("PERSON.SECURITY.CERTIFICATE.OUT");
        try {
            getData(PERSON_SECURITY_CERTIFICATE, PERSON_SECURITY_CERTIFICATE_OUT, "FID", ri, 0, "ag_ry");
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }

    public void startSynDataTz() {

        DataApi ri = new DataApi(DBConfig.readPop("daq.task.pconnect.url"));
        String PERSON_SECURITY_CERTIFICATE = DBConfig.readPop("PERSON.SECURITY.CERTIFICATE");
        String PERSON_SPECIAL_CERTIFICATE_OUT = DBConfig.readPop("PERSON.SPECIAL.CERTIFICATE.OUT");
        try {
            getData(PERSON_SECURITY_CERTIFICATE, PERSON_SPECIAL_CERTIFICATE_OUT, "FID", ri, 0, "tz_zy");
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }

    public void startDelete(String type) {

        DataApi ri = new DataApi(DBConfig.readPop("daq.task.pconnect.url"));
        String tabName = "";
        String tabNameOut = "";
        if ("ag_ry".equals(type)) {
            tabName = DBConfig.readPop("PERSON.SECURITY.CERTIFICATE");
            tabNameOut = DBConfig.readPop("PERSON.SECURITY.CERTIFICATE.OUT");
        }
        if ("axz".equals(type)) {
            tabName = DBConfig.readPop("COMPANY.SAFETY.CERTIFICATE");
            tabNameOut = DBConfig.readPop("COMPANY.SAFETY.CERTIFICATE.OUT");
        }
        if("tz_zy".equals(type)){
            tabName = DBConfig.readPop("PERSON.SECURITY.CERTIFICATE");
            tabNameOut = DBConfig.readPop("PERSON.SPECIAL.CERTIFICATE.OUT");
        }

        try {
            dataCompareDel(tabName, tabNameOut, "FID", ri, 0, type);
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }


    private String getObjType(Object obj) {

        String type = "";
        if (obj instanceof Integer) {
            type = "int";
        } else if (obj instanceof String) {
            type = "String";
        } else if (obj instanceof Double) {
            type = "Double";
        } else if (obj instanceof Float) {
            type = "Float";
        } else if (obj instanceof Long) {
            type = "Long";
        }
        return type;
    }

    private Map<String, Object> cloneMap(Map<String, Object> resMap) {
        Map<String, Object> cloneMap = new HashMap<>();
        for (Map.Entry<String, Object> map : resMap.entrySet()) {
            Object key = map.getKey();
            Object value = map.getValue();
            cloneMap.put(key + "", value);
        }
        return cloneMap;
    }


    private int cacheData(String outTabName, int outCount, DBQueryUtils dbQueryUtils, String key, int b, int d) {

        if (b == 3) {
            return -1;
        }
        int num = 5000;
        int batch = (outCount / num) + (outCount % num != 0 ? 1 : 0);
        int count = 0;
        for (int i = 0; i < batch; i++) {
            PageInfo pageInfo = getPageInfo(outTabName, outCount, dbQueryUtils, key, i + 1, num, 0);
            if (XaUtil.isEmpty(pageInfo)) {
                return -1;
            }
        }

        Set<String> keys = redisOperator.keys(outTabName + "*");
        if (keys.size() != outCount) {
            keys.clear();
            b += 1;
            cacheData(outTabName, outCount, dbQueryUtils, key, b, d);
        }
        return count;
    }

    private PageInfo getPageInfo(String outTabName, int outCount, DBQueryUtils dbQueryUtils, String key, int page, int num, int d) {
        PageInfo pageInfo = null;
        try {
            pageInfo = dbQueryUtils.queryTableDataByPaging(dbName, outTabName, page, num, "select * from " + outTabName);
            List<Map> resultlist = pageInfo.getResultList();
            for (Map resMap : resultlist) {
                String keyValue = resMap.get(key) + "";
                resMap.remove("FTXT1");
                resMap.remove("FTXT2");
                resMap.remove("FTXT3");
                resMap.remove("FTXT4");
                resMap.remove("FTXT5");
                JSONObject json = JSONObject.fromObject(resMap);
                jedisCluster.set(outTabName + "_" + keyValue, json.toString());
            }
        } catch (Exception e) {
            e.printStackTrace();
            if (3 == d) {
                return null;
            }
            d += 1;
            pageInfo = getPageInfo(outTabName, outCount, dbQueryUtils, key, page, num, d);
            if (XaUtil.isEmpty(pageInfo)) {
                pageInfo = getPageInfo(outTabName, outCount, dbQueryUtils, key, page, num, d);
            }
        }
        return pageInfo;
    }




    private void errorRecord(String json, String resCode, String tableName, String task) {
        String url = "";
        ConstConfig.DBType dbType = ConstConfig.DBType.valueOf(Integer.valueOf(recordDbSType));
        if ("3".equals(recordDbSType)) {
            url = "jdbc:mysql://" + recordIp + ":" + recordPort + "/" + recordDbName + "?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useInformationSchema=true&useSSL=false";
        } else {
            url = DBInfoUtil.getDBUrl(dbType, recordIp, recordPort, recordDbName);
        }

        DBQueryUtils dbQueryUtilsMysql = new DBQueryUtils(DBInfoUtil.getDBDriverName(dbType),
                url, recordUserName, recordPassword, 5);

        List<Map<String, Object>> dataList = new ArrayList<>();
        Map<String, Object> map = new HashMap<>();
        map.put("id", IdGenerator.createNewId());
        map.put("task_id", task);
        map.put("time", CommonUtils.YYYYMMDDHHMMSS.format(new Date()));
        map.put("error_message", resCode);
        map.put("error_data", json);
        map.put("table_name", tableName);
        dataList.add(map);
        try {
            dbQueryUtilsMysql.insertBatchData("data_exchange_error", recordDbName, dataList);
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }
}
